/**
 * Created by lzfto
 * Created time: 2025/05/25
 */
package org.lzfto.flink.demo.table.service.impl;

import static org.apache.flink.table.api.Expressions.$;

import java.util.concurrent.ExecutionException;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.lzfto.flink.demo.common.table.PhysicalBasicTable;
import org.lzfto.flink.demo.table.service.FlinkConnectorTestService;

/**
 * 从本地测试库的一个schema插入到另外一个
 * 192.168.1.246 test/1232018
 * 采用的是批处理模式
 */
public class CustomJdbcConnectorTest  implements FlinkConnectorTestService {
    @Override
    public void run() throws Exception {
        //1. 创建执行环境
        EnvironmentSettings env = EnvironmentSettings.newInstance().inBatchMode()
        		.withConfiguration(createLowLevelConfig())
        		.build();
        TableEnvironment tabEnv = TableEnvironment.create(env);
        //2. 创建表源头表和目标表
        String sourceWith=PhysicalBasicTable.getSourceTable("student").withConnector("jdbc").build();
        String sourceTableDDL = """
                create table student(
                   student_id BIGINT,
                   name String,
                   gender String,
                   birth_date Date,
                   email String
                )
                """+sourceWith;
        tabEnv.executeSql(sourceTableDDL);

        //目标表
        //Flink 中 字段类型需要严格匹配，否则会出现许多奇怪的问题
        //很可惜内置的函数不能用于create table语法的字段定义部分
        String targetWith=PhysicalBasicTable.getTargetTable("student")
        		.withConnector("custom-jdbc")
        		.build();
        String targetTableDDL = """
                create table student_target(
                   student_id BIGINT,
                   name STRING,
                   gender STRING,
                   birth_date DATE,
                   email STRING
                )
                """+targetWith;
        tabEnv.executeSql(targetTableDDL);

        //3.插入前，请清理目标表数据
        //TRUNCATE TABLE statement is not supported in streaming mode
        //使用自定义连接器的删除，但是由于太低效了，删除了吧！
        TableResult delResult =tabEnv.executeSql("delete from student_target");
        try {
			delResult.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}

        //4.执行插入操作
        //主要2000年之后出生
        Table srcTable = tabEnv.sqlQuery("SELECT * FROM student WHERE birth_date >= DATE '2000-01-01'");
        //只要男生
        srcTable=srcTable.filter($("gender").isEqual("Male"));
        TableResult insResult=srcTable.executeInsert("student_target");
        try {
			insResult.await();
            System.out.println(insResult.getResultKind().name());
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		} catch (Exception e) {
            e.printStackTrace();
        }

        System.out.println("执行完成");
    }
}
